perf(aggregate): consecutive-keys (last-group) cache for GROUP BY on clustered data - #23986
Draft
zhuqi-lucas wants to merge 2 commits into
Draft
perf(aggregate): consecutive-keys (last-group) cache for GROUP BY on clustered data#23986zhuqi-lucas wants to merge 2 commits into
zhuqi-lucas wants to merge 2 commits into
Conversation
…n GROUP BY Real-world data frequently arrives with runs of identical group keys (time-ordered logs, clustered writes). Today `GroupValues::intern` pays the full hash + hash-table-probe cost for every row, even when a row's key equals the previous row's — and for high-cardinality keys the table is far too large to stay cache-resident, so each probe is typically an L3/DRAM round trip. Remember the previous row's key and its group/payload; when the current row matches, reuse the answer and skip the probe (and for primitives, the hash as well). A hit replaces a random memory access with one or two well-predicted compares against L1-resident state; a miss costs a single compare (<1ns). Break-even hit rate is under 1%. This is the same optimization ClickHouse ships as its "consecutive keys optimization" (`LastElementCache` in ColumnsHashing, refined in ClickHouse#57872). Measured run lengths on ClickBench hits.parquet (file order, which is what the Partial aggregation phase sees): UserID avg run 10.1 (90.1% hit rate), SearchPhrase 11.7 (91.5%), RegionID 11.0 (90.9%), URL 1.4 (30.2%). Three implementations, one per single-column path: - `GroupValuesPrimitive` (primitive keys): cache `(last_key, last_group)` fields; the guard compares the canonicalized value itself, so a hit skips the hash too. Invalidated in `emit` / `clear_shrink`, since both reassign group indices. - `ArrowBytesMap::insert_if_new_inner` (Utf8 / LargeUtf8 / Binary via `GroupValuesBytes`, also used by distinct-aggregate accumulators): compare the current row's bytes against the previous input row's — adjacent rows in the same values buffer, so the comparison is cache-hot. Per-call state, no invalidation concerns. - `ArrowBytesViewMap::insert_if_new_inner` (Utf8View / BinaryView via `GroupValuesBytesView`): for inline views (len <= 12) the u128 view comparison is a complete equality check; longer values get length + 4-byte-prefix guards from the views before touching bytes. Intervening nulls do not invalidate the cached mapping (a value's group/payload never changes within an accumulation), and the make-payload-once / observe-per-row-in-order contract of the maps is preserved by the fast path. ClickBench (hits.parquet, 5 iterations, release-nonlto, both binaries built from this commit's base): Q33 (URL) 3621ms -> 2969ms (-18.0%) Q15 (UserID) 304ms -> 271ms (-10.7%) Q12 (SearchPhrase) 389ms -> 357ms ( -8.2%) Q7 (AdvEngineID) 20ms -> 18ms ( -6.2%) Q27 (CounterID) 739ms -> 704ms ( -4.8%) The win scales with (hit rate x probe cost): Q33's URL table is the largest string hash table, so even a 30% hit rate pays the most per skipped probe; Q27's ~6K-entry table is L1/L2-resident, so gains are small even at ~100% hit rate. Tests: consecutive-run + interleaved-null payload-assignment tests for both maps, and a primitive test covering the emit(First(n)) cache invalidation (a stale cache would assign shifted group ids).
…n GROUP BY Extends the last-group cache to `GroupValuesColumn` (both the vectorized and the streaming/scalarized intern), covering multi-column group keys such as ClickBench's `GROUP BY "UserID", "SearchPhrase"`. Multi-column runs need every column to match its predecessor, so instead of a per-row cached slot the batch computes an adjacent-equality mask up front, guarded by a two-tier gate: - A single sequential scan over the (already computed) hashes counts adjacent hash matches. Hash equality is a necessary condition for row equality, so a shuffled batch — e.g. the Final phase after hash repartitioning, where runs cannot survive — fails the gate and pays nothing beyond this one scan. Correctness never rests on the hashes: they only decide whether the mask is worth building. - Run-heavy batches build the precise mask: one vectorized `not_distinct` pass per column, AND-ed. `not_distinct` treats `null == null` as equal, matching GROUP BY semantics, and never returns nulls. Column types it does not support (e.g. nested types handled by `RowsGroupColumn`) make the helper return `None`, falling back to the normal path. Integration is deliberately non-invasive to the vectorized machinery: masked rows are skipped in `collect_vectorized_process_context` (they enter none of the append / equal-to / remaining lists), and a final ascending fill pass assigns each one its predecessor's group index — forward propagation resolves multi-row runs from the run head, which took the normal path. The streaming path checks the mask at the top of its row loop and reuses the previous row's group index directly; sorted input has perfect runs, so it benefits the most. ClickBench multi-column queries (hits.parquet, 5-8 iterations, release-nonlto, same-commit baseline): Q18 (UserID, m, SearchPhrase) 3714ms -> 3312ms (-10.8%) Q32 (WatchID, ClientIP) 3478ms -> 3122ms (-10.2%, 8 iters) Q17 (UserID, SearchPhrase) 740ms -> 700ms ( -5.4%) Q16 (UserID, SearchPhrase) 819ms -> 795ms ( -3.0%) Tests: a multi-column runs test asserting identical group assignments on both intern paths — including the "one column runs, the other breaks the run" case and null==null runs — plus a no-runs batch exercising the gate fallback. Probe-verified (temporary eprintln) that the fast path fires for exactly the designed rows on both paths and that the gate rejects the no-runs batch.
Contributor
Author
|
run benchmark clickbench_partitioned |
github-actions
Bot
added
physical-expr
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing perf/agg-last-group-cache (4c05353) to 68d5874 (merge-base) diff using: clickbench_partitioned File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: CPU Details (lscpu)Details
Resource Usageclickbench_partitioned — base (merge-base)
clickbench_partitioned — branch
File an issue against this benchmark runner |
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## main #23986 +/- ##
========================================
Coverage 80.75% 80.76%
========================================
Files 1096 1096
Lines 373282 373798 +516
Branches 373282 373798 +516
========================================
+ Hits 301440 301885 +445
- Misses 53869 53906 +37
- Partials 17973 18007 +34 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
3 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Which issue does this PR close?
Rationale for this change
GroupValues::internpays the full hash + hash-table-probe cost for every input row, even when a row's group key equals the previous row's. For high-cardinality keys the table is far too large to stay cache-resident, so each probe is typically an L3/DRAM round trip.Real-world data very often arrives with runs of identical keys (time-ordered logs, clustered writes). Measured on ClickBench
hits.parquetin file order — which is what the Partial aggregation phase sees, since it runs before repartitioning:Remembering the previous row's key and group index lets a run reuse the answer: a hit replaces a random memory access with one or two well-predicted compares against L1-resident state; a miss costs a single compare. Break-even hit rate is under 1%. This is the same optimization ClickHouse ships as its consecutive keys optimization (
LastElementCacheinColumnsHashing, refined in ClickHouse#57872).The win scales with
hit rate × probe cost: Q33's URL table is the largest string hash table, so even a 30% hit rate pays the most per skipped probe, while Q27's ~6K-entry CounterID table is L1/L2-resident and gains little at ~100% hit rate.What changes are included in this PR?
Four implementations, one per intern path:
GroupValuesPrimitive(single primitive key): cache(last_key, last_group)fields; the guard compares the canonicalized value itself, so a hit skips the hash as well as the probe. Invalidated inemit/clear_shrinksince both reassign group indices.ArrowBytesMap::insert_if_new_inner(singleUtf8/LargeUtf8/Binarykey viaGroupValuesBytes; also used by distinct-aggregate accumulators, which benefit for free): compare the current row's bytes against the previous input row's — adjacent rows in the same values buffer, so the comparison is cache-hot. Per-call state, no invalidation concerns.ArrowBytesViewMap::insert_if_new_inner(singleUtf8View/BinaryViewkey viaGroupValuesBytesView): for inline views (len ≤ 12) the u128 view comparison is a complete equality check; longer values get length + 4-byte-prefix guards from the views before touching bytes.GroupValuesColumn(multi-column keys, both the vectorized and streaming interns): a batch-level adjacent-equality mask (not_distinctper column, AND-ed —null == nullcounts as equal, matching GROUP BY semantics), guarded by a two-tier gate: a sequential scan over the already-computed hashes counts adjacent hash matches, and only run-heavy batches (≥ ~6%) build the precise mask. Shuffled inputs — e.g. the Final phase after hash repartitioning — fail the gate and pay nothing beyond that one scan. Correctness never rests on the hashes; they only decide whether the mask is worth building. Nested-type columns (handled byRowsGroupColumn) make the helper returnNoneand fall back to the normal path.Integration is deliberately non-invasive to the vectorized machinery: masked rows are skipped in
collect_vectorized_process_context(they enter none of the append / equal-to / remaining lists), and a final ascending fill pass assigns each one its predecessor's group index.Intervening nulls do not invalidate any of the caches (a value's group never changes within an accumulation), and the make-payload-once / observe-per-row-in-order contract of the byte maps is preserved.
Are these changes tested?
consecutive_keys_runs_and_emit_invalidation(primitive): runs with interleaved nulls, plus theemit(First(n))index-shift trap — a stale cache would assign shifted group ids.string_map_consecutive_keys_runs/view_map_consecutive_keys_runs: payload-assignment equality between run and scattered occurrences, interleaved nulls, inline and non-inline string lengths,make_payload_fncalled exactly once per distinct.test_intern_consecutive_keys_multi_column: identical group assignments on both intern paths, including the "one column runs while the other breaks the run" case andnull == nullruns;test_intern_consecutive_keys_no_runsexercises the gate fallback.eprintln) that the fast path fires for exactly the designed rows on both multi-column paths and that the gate rejects the no-runs batch.aggregates::suite (158),datafusion-physical-expr-commonsuite (77), and workspace clippy with-D warningsall pass.Are there any user-facing changes?
No API changes. Local ClickBench numbers (hits.parquet, 5–8 iterations,
release-nonlto, macOS ARM, both binaries built from this branch's base commit) — CI benchmarks to confirm:No regressions observed; queries whose plans don't touch these paths are unchanged.